{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MhoQ0WE77laV"
      },
      "source": [
        "##### Copyright 2020 The TensorFlow Authors."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "_ckMIh7O7s6D"
      },
      "outputs": [],
      "source": [
        "#@title Licensed under the Apache License, Version 2.0 (the \"License\");\n",
        "# you may not use this file except in compliance with the License.\n",
        "# You may obtain a copy of the License at\n",
        "#\n",
        "# https://www.apache.org/licenses/LICENSE-2.0\n",
        "#\n",
        "# Unless required by applicable law or agreed to in writing, software\n",
        "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
        "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
        "# See the License for the specific language governing permissions and\n",
        "# limitations under the License."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jYysdyb-CaWM"
      },
      "source": [
        "# Распределенный ввод"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "S5Uhzt6vVIB2"
      },
      "source": [
        "<table class=\"tfo-notebook-buttons\" align=\"left\">\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://www.tensorflow.org/tutorials/distribute/input\"><img src=\"https://www.tensorflow.org/images/tf_logo_32px.png\" />Смотрите на TensorFlow.org</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://colab.research.google.com/github/tensorflow/docs-l10n/blob/master/site/ru/tutorials/distribute/input.ipynb\"><img src=\"https://www.tensorflow.org/images/colab_logo_32px.png\" />Запустите в Google Colab</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a target=\"_blank\" href=\"https://github.com/tensorflow/docs-l10n/blob/master/site/ru/tutorials/distribute/input.ipynb\"><img src=\"https://www.tensorflow.org/images/GitHub-Mark-32px.png\" />Изучайте код на GitHub</a>\n",
        "  </td>\n",
        "  <td>\n",
        "    <a href=\"https://storage.googleapis.com/tensorflow_docs/docs-l10n/site/ru/tutorials/distribute/input.ipynb\"><img src=\"https://www.tensorflow.org/images/download_logo_32px.png\" />Скачайте ноутбук</a>\n",
        "  </td>\n",
        "</table>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "8a2322303a3f"
      },
      "source": [
        "Note: Вся информация в этом разделе переведена с помощью русскоговорящего Tensorflow сообщества на общественных началах. Поскольку этот перевод не является официальным, мы не гарантируем что он на 100% аккуратен и соответствует [официальной документации на английском языке](https://www.tensorflow.org/?hl=en). Если у вас есть предложение как исправить этот перевод, мы будем очень рады увидеть pull request в [tensorflow/docs](https://github.com/tensorflow/docs) репозиторий GitHub. Если вы хотите помочь сделать документацию по Tensorflow лучше (сделать сам перевод или проверить перевод подготовленный кем-то другим), напишите нам на [docs-ru@tensorflow.org list](https://groups.google.com/a/tensorflow.org/forum/#!forum/docs-ru)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FbVhjPpzn6BM"
      },
      "source": [
        "API [tf.distribute](https://www.tensorflow.org/guide/distributed_training) предоставляет пользователям простой способ масштабировать обучение модели с одного устройства на несколько. При масштабировании своей модели пользователи также должны распределять входные данные между несколькими устройствами. `tf.distribute` предоставляет API-интерфейс, с помощью которого вы можете автоматически распределять входные данные между устройствами.\n",
        "\n",
        "Это руководство покажет вам различные способы создания распределенных датасетов и итераторов с помощью `tf.distribute`. Дополнительно будут рассмотрены следующие темы:\n",
        "- Параметры использования, сегментирования и пакетной обработки при использовании `tf.distribute.Strategy.experimental_distribute_dataset` и `tf.distribute.Strategy.experimental_distribute_datasets_from_function.\n",
        "- Различные способы итерации по распределенному набору данных.\n",
        "- Различия между API-интерфейсами `tf.distribute.Strategy.experimental_distribute_dataset`/`tf.distribute.Strategy.experimental_distribute_datasets_from_function` и `tf.data`, а также любые ограничения, с которыми пользователи могут столкнуться при их использовании.\n",
        "\n",
        "В этом руководстве не рассматривается использование распределенного ввода с помощью Keras API ."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "MM6W__qraV55"
      },
      "source": [
        "## Распределенные датасеты"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lNy9GxjSlMKQ"
      },
      "source": [
        "Для масштабирования данных с помощью API `tf.distribute` рекомендуется использовать объект `tf.data.Dataset`. `tf.distribute` был создан для эффективной работы с `tf.data.Dataset`(например, автоматическая предварительная выборка данных на каждое устройство-ускоритель), при этом в реализацию регулярно включаются оптимизации производительности. Если у вас есть вариант использования чего-то другого, кроме `tf.data.Dataset`, обратитесь к\n",
        "[разделу](#tensorinputs) в этом руководстве.\n",
        "В обычном(нераспределенном) цикле обучения, пользователи сначала создают экземпляр `tf.data.Dataset`, а затем перебирают элементы. Например:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "pCu2Jj-21AEf"
      },
      "outputs": [],
      "source": [
        "import tensorflow as tf\n",
        "\n",
        "# Helper libraries\n",
        "import numpy as np\n",
        "import os\n",
        "\n",
        "print(tf.__version__)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "dzLKpmZICaWN"
      },
      "outputs": [],
      "source": [
        "global_batch_size = 16\n",
        "# Создание объекта tf.data.Dataset.\n",
        "dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(global_batch_size)\n",
        "\n",
        "@tf.function\n",
        "def train_step(inputs):\n",
        "  features, labels = inputs\n",
        "  return labels - 0.3 * features\n",
        "\n",
        "# Итерация по датасету с помощью конструкции for..in...\n",
        "for inputs in dataset:\n",
        "  print(train_step(inputs))\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ihrhYDYRrVLH"
      },
      "source": [
        "Чтобы пользователи могли использовать стратегию `tf.distribute` с минимальными изменениями в существующем коде, были введены два API, которые будут расперделять экземпляр `tf.data.Dataset` и возвращать объект распределенного набора данных. Затем пользователь может итерироваться по распределенному датасету и обучать свою модель, как и раньше. Давайте рассмотрим оба API - `tf.distribute.Strategy.experimental_distribute_dataset` и `tf.distribute.Strategy.experimental_distribute_datasets_from_function` более подробно:"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "4AXoHhrsbdF3"
      },
      "source": [
        "### `tf.distribute.Strategy.experimental_distribute_dataset`"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "5mVuLZhbem8d"
      },
      "source": [
        "#### Использование\n",
        "\n",
        "Этот API принимает в качестве входных данных экземпляр `tf.data.Dataset` и возвращает экземпляр `tf.distribute.DistributedDataset`. Вы должны оперделить размер пакета(batch) `tf.data.Dataset` равным глобальному размеру пакета. Глобальный размер пакета - это количество строк данных, которые вы хотите обработать на всех устройствах за 1 шаг. Вы можете итерироваться по распределенному датасету в стиле Python или создать итератор с помощью `iter`. Возвращенный объект не является экземпляром `tf.data.Dataset` и не поддерживает никаких других API, которые каким-либо образом преобразовывают или проверяют набор данных.\n",
        "Это рекомендуемый API, если у вас нет своего особого способа распределния входных данных по разным устройствам.\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "F2VeZUWUj5S4"
      },
      "outputs": [],
      "source": [
        "global_batch_size = 16\n",
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "\n",
        "dataset = tf.data.Dataset.from_tensors(([1.], [1.])).repeat(100).batch(global_batch_size)\n",
        "# Распределяем входные данные с помощью `experimental_distribute_dataset`.\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)\n",
        "# 1 глобальный пакет данных, загружаемый в модель за 1 шаг.\n",
        "print(next(iter(dist_dataset)))"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "QPceDmRht54F"
      },
      "source": [
        "#### Свойства"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0Qb6nDgxiN_n"
      },
      "source": [
        "#####  Разделение на пакеты(Batching)\n",
        "\n",
        "`tf.distribute` rebatches the input `tf.data.Dataset` instance with a new batch size that is equal to the global batch size divided by the number of replicas in sync. The number of replicas in sync is equal to the number of devices that are taking part in the gradient allreduce during training. When a user calls `next` on the distributed iterator, a per replica batch size of data is returned on each replica. The rebatched dataset cardinality will always be a multiple of the number of replicas. Here are a couple of\n",
        "examples:\n",
        "\n",
        "`tf.distribute` обновляет входной экземпляр `tf.data.Dataset`, устанавливая новый размер пакета, равный глобальному размеру пакета, деленному на количество синхронизируемых реплик. Количество синхронизируемых реплик равно количеству устройств, участвующих в уменьшении градиента во время обучения. Когда пользователь вызывает `next` на итераторе распределнного объекта, размер пакета данных для каждой реплики возвращается на каждой реплике. Количество элементов повторно собранного набора данных всегда будет кратным количеству реплик. Вот несколько примеров:\n",
        "\n",
        "* `tf.data.Dataset.range(6).batch(4, drop_remainder=False)`\n",
        "  * Без распределения:\n",
        "    * Batch 1: [0, 1, 2, 3]\n",
        "    * Batch 2: [4, 5]\n",
        "  * С распределением на 2 реплики.\n",
        "    Последний батч ([4, 5]) также распределяется между 2-мя реплтками.\n",
        "\n",
        "    * Batch 1:\n",
        "      * Replica 1:[0, 1]\n",
        "      * Replica 2:[2, 3]\n",
        "    * Batch 2:\n",
        "      * Replica 2: [4]\n",
        "      * Replica 2: [5]\n",
        "\n",
        "\n",
        "\n",
        "* `tf.data.Dataset.range(4).batch(4)`\n",
        "  * Без распределения:\n",
        "    * Batch 1: [[0], [1], [2], [3]]\n",
        "  * С распределением на 5 реплик:\n",
        "    * Batch 1:\n",
        "      * Replica 1: [0]\n",
        "      * Replica 2: [1]\n",
        "      * Replica 3: [2]\n",
        "      * Replica 4: [3]\n",
        "      * Replica 5: []\n",
        "\n",
        "* `tf.data.Dataset.range(8).batch(4)`\n",
        "  * Без распределения:\n",
        "    * Batch 1: [0, 1, 2, 3]\n",
        "    * Batch 2: [4, 5, 6, 7]\n",
        "  * С распределением на 3 реплики:\n",
        "    * Batch 1:\n",
        "      * Replica 1: [0, 1]\n",
        "      * Replica 2: [2, 3]\n",
        "      * Replica 3: []\n",
        "    * Batch 2:\n",
        "      * Replica 1: [4, 5]\n",
        "      * Replica 2: [6, 7]\n",
        "      * Replica 3: []\n",
        "\n",
        "Примечание. Приведенные выше примеры только иллюстрируют, как глобальный пакет разделяется на разные реплики. Не рекомендуется полагаться на фактические значения, которые могут быть на каждой реплике, поскольку они могут меняться в зависимости от реализации.\n",
        "\n",
        "Повторное объединение набора данных имеет пространственную сложность, которая линейно увеличивается с количеством реплик. Это значит, что в случае использования обучения с несколькими устройствами конвейер ввода может столкнуться с ошибками нехватки памяти."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IszBuubdtydp"
      },
      "source": [
        "#####  Масштабирование(Sharding)\n",
        "\n",
        "`tf.distribute` также выполняет автоматическое масштабирование входного набора данных при распределенном обучении. Каждый набор данных создается на центральном процессоре устройства-воркера. Автомасштабирование набора данных по набору воркеров означает, что каждому воркеру назначается подмножество всего набора данных(если установлена правильная политика масштабирования - `tf.data.experimental.AutoShardPolicy`). Это необходимо для обеспечения того, чтобы на каждом шаге каждый воркер обрабатывал пакет неперекрывающихся элементов набора данных равный размеру глобального пакета. Автомасштабирование имеет несколько различных опций, которые можно указать с помощью `tf.data.experimental.DistributeOptions`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "jwJtsCQhHK-E"
      },
      "outputs": [],
      "source": [
        "dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(64).batch(16)\n",
        "options = tf.data.Options()\n",
        "options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.DATA\n",
        "dataset = dataset.with_options(options)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "J7fj3GskHC8g"
      },
      "source": [
        "Существует три разных параметра, которые вы можете установить для `tf.data.experimental.AutoShardPolicy`:\n",
        "\n",
        "* AUTO: это параметр по умолчанию, который означает, что будет предпринята попытка разбить данные типа FILE. Попытка разбиения с помощью политики FILE завершается ошибкой, если не обнаружен файловый набор данных. После этого `tf.distribute` вернется к сегментированию по DATA. Обратите внимание, что если входной набор данных основан на файлах, но количество файлов меньше количества воркеров, будет вызвана ошибка `InvalidArgumentError`. Если это произойдет, явно установите для политики значение `AutoShardPolicy.DATA` или разделите источник ввода на файлы меньшего размера, чтобы количество файлов было больше, чем количество рабочих процессов.\n",
        "* FILE: выберите это вариант, если хотите разделить входные файлы на всех воркеров. Вы должны использовать эту опцию, если количество входных файлов намного больше, чем количество рабочих, и данные в файлах распределены равномерно. Обратной стороной этого варианта является наличие простаивающих рабочих, если данные в файлах распределены неравномерно. Если количество файлов меньше, чем количество рабочих, будет вызвана ошибка `InvalidArgumentError`. Если это произойдет, явно установите политику на `AutoShardPolicy.DATA`.\n",
        "\n",
        "В качестве примера давайте распределим 2 файла по 2 воркерам и 2 репликам, по 1 реплике на каждого воркера. Файл 1 содержит [0, 1, 2, 3, 4, 5] и Файл 2 содержит [6, 7, 8, 9, 10, 11]. Пусть общее количество синхронизируемых реплик будет 2, а размер глобального пакета - 4.\n",
        "\n",
        "  * Worker 0:\n",
        "    * Batch 1 =  Replica 1: [0, 1]\n",
        "    * Batch 2 =  Replica 1: [2, 3]\n",
        "    * Batch 3 =  Replica 1: [4]\n",
        "    * Batch 4 =  Replica 1: [5]\n",
        "  * Worker 1:\n",
        "    * Batch 1 =  Replica 2: [6, 7]\n",
        "    * Batch 2 =  Replica 2: [8, 9]\n",
        "    * Batch 3 =  Replica 2: [10]\n",
        "    * Batch 4 =  Replica 2: [11]\n",
        "\n",
        "* DATA: Эта политика автомасштабирует элементы для всех воркеров. Каждый из воркеров будет читать весь набор данных и обрабатывать только назначенный ему сегмент. Все остальные сегменты данных будут отброшены. Обычно это используется, если количество входных файлов меньше, чем количество рабочих процессов, и вы хотите улучшить распределение данных по всем рабочим процессам. Обратной стороной является то, что весь набор данных будет прочитан каждым воркером.\n",
        "\n",
        "В качестве примера давайте распределим 1 файл по 2 воркерам. Файл 1 содержит [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Пусть общее количество синхронизируемых реплик равно 2.\n",
        "\n",
        "  * Worker 0:\n",
        "    * Batch 1 =  Replica 1: [0, 1]\n",
        "    * Batch 2 =  Replica 1: [4, 5]\n",
        "    * Batch 3 =  Replica 1: [8, 9]\n",
        "  * Worker 1:\n",
        "    * Batch 1 =  Replica 2: [2, 3]\n",
        "    * Batch 2 =  Replica 2: [6, 7]\n",
        "    * Batch 3 =  Replica 2: [10, 11]\n",
        "\n",
        "* OFF: Если отключить автомасштабирование, то каждый воркер будет обрабатывать все данные.\n",
        "\n",
        "Давайте распределим 1 файл по 2 воркерам. Файл 1 содержит [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]. Пусть общее количество синхронизируемых реплик равно 2. Тогда каждый воркер увидит следующее распределение:\n",
        "\n",
        "  * Worker 0:\n",
        "    * Batch 1 =  Replica 1: [0, 1]\n",
        "    * Batch 2 =  Replica 1: [2, 3]\n",
        "    * Batch 3 =  Replica 1: [4, 5]\n",
        "    * Batch 4 =  Replica 1: [6, 7]\n",
        "    * Batch 5 =  Replica 1: [8, 9]\n",
        "    * Batch 6 =  Replica 1: [10, 11]\n",
        "\n",
        "  * Worker 1:\n",
        "    * Batch 1 =  Replica 2: [0, 1]\n",
        "    * Batch 2 =  Replica 2: [2, 3]\n",
        "    * Batch 3 =  Replica 2: [4, 5]\n",
        "    * Batch 4 =  Replica 2: [6, 7]\n",
        "    * Batch 5 =  Replica 2: [8, 9]\n",
        "    * Batch 6 =  Replica 2: [10, 11] "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "OK46ZJGPH5H2"
      },
      "source": [
        "##### Предварительная загрузка\n",
        "\n",
        "По умолчанию `tf.distribute` добавляет вызов метода предварительной выборки в конец экземпляра `tf.data.Dataset`. Аргумент преобразования предварительной выборки, который равен `buffer_size`, равен количеству синхронизируемых реплик."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "PjiGSY3gtr6_"
      },
      "source": [
        "### `tf.distribute.Strategy.experimental_distribute_datasets_from_function`"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "bAXAo_wWbWSb"
      },
      "source": [
        "#### Использование\n",
        "\n",
        "Этот API принимает функцию ввода и возвращает экземпляр `tf.distribute.DistributedDataset`. Функция ввода, которую передает пользователь, имеет аргумент `tf.distribute.InputContext` и должна возвращать экземпляр `tf.data.Dataset`. API `tf.distribute` не вносит никаких изменений в пользовательский экземпляр `tf.data.Dataset`, возвращенный функцией ввода. Пользователь несет ответственность за пакетирование и сегментирование набора данных. `tf.distribute` только вызывает переданную функцию ввода на устройстве каждого из воркеров. Помимо того, что пользователи могут указывать свою собственную логику пакетирования и сегментирования, этот API показывает лучшую масштабируемость и производительность по сравнению с `tf.distribute.Strategy.experimental_distribute_dataset`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "9ODch-OFCaW4"
      },
      "outputs": [],
      "source": [
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "\n",
        "def dataset_fn(input_context):\n",
        "  batch_size = input_context.get_per_replica_batch_size(global_batch_size)\n",
        "  dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(64).batch(16)\n",
        "  dataset = dataset.shard(\n",
        "    input_context.num_input_pipelines, input_context.input_pipeline_id)\n",
        "  dataset = dataset.batch(batch_size)\n",
        "  dataset = dataset.prefetch(2) # Предварительная загрузка 2 пакетов на каждом устройстве.\n",
        "  return dataset\n",
        "\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_datasets_from_function(dataset_fn)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "M1bpzPYzt_R7"
      },
      "source": [
        "#### Свойства"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7cgzhwiiuBvO"
      },
      "source": [
        "##### Разделение на пакеты(Batching)\n",
        "\n",
        "Экземпляр `tf.data.Dataset`, который возвращает функция ввода, должен быть упакован с использованием размера пакета для каждой реплики. Размер пакета для каждой реплики - это глобальный размер пакета, разделенный на количество реплик, участвующих в распределенном обучении. Так как `tf.distribute` вызывает функцию ввода на устройстве ЦП каждого из воркеров, набор данных, созданный для данного воркера, должен быть готов к использованию всеми репликами на этом воркере."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "e-wlFFZbP33n"
      },
      "source": [
        "##### Масштабирование(Sharding)\n",
        "\n",
        "Объект `tf.distribute.InputContext`, который неявно передается в качестве аргумента функции ввода пользователя, создается `tf.distribute` под капотом. Он содержит информацию о количестве воркеров, текущем идентификаторе рабочего процесса и т.п. Функция ввода может обрабатывать масштабирование в соответствии с политиками, установленными пользователем в свойствах объекта `tf.distribute.InputContext`.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7TGwnDM-ICHf"
      },
      "source": [
        "##### Предварительная загрузка\n",
        "\n",
        "`tf.distribute` **не** добавляет метод предварительной выборки в конец `tf.data.Dataset`, возвращаемого функцией ввода, предоставленной пользователем."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iOMsf8kyZZpv"
      },
      "source": [
        "Примечание:\n",
        "Оба метода - `tf.distribute.Strategy.experimental_distribute_dataset` и `tf.distribute.Strategy.experimental_distribute_datasets_from_function` возвращают **экземпляры `tf.distribute.DistributedDataset`, которые не относятся к типу `tf.data.Dataset`**. Вы можете итерироваться по этим объектам(как показано в разделе «Распределенные итераторы») и использовать свойство `element_spec`."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dL3XbI1gzEjO"
      },
      "source": [
        "## Распределенные итераторы"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "w8y54-o9T2Ni"
      },
      "source": [
        "Подобно нераспределенным объектам `tf.data.Dataset`, вам нужно будет создать итератор для объекта `tf.distribute.DistributedDataset`, чтобы итерироваться по нему и получать доступ к элементам.\n",
        "Ниже приведены способы, которыми вы можете создать `tf.distribute.DistributedIterator` и использовать его в процессе обучения вашей модели:\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "FlKh8NV0uOtZ"
      },
      "source": [
        "### Использование"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "eSZz6EqOuSlB"
      },
      "source": [
        "#### Используйте Python конструкцию цикла for \n",
        "\n",
        "Вы можете использовать циклы Python для итерации по `tf.distribute.DistributedDataset`. Элементы, возвращаемые объектом `tf.distribute.DistributedIterator`, могут быть `tf.Tensor` или `tf.distribute.DistributedValues`, содержащие значения для каждой реплики. Размещение цикла внутри `tf.function` даст прирост производительности. Однако в настоящее время `break` и `return` не поддерживаются для циклов по `tf.distribute.DistributedDataset`, который помещается внутри `tf.function`. \n",
        "Мы также не поддерживаем размещение цикла внутри `tf.function` при использовании многопользовательских стратегий, таких как `tf.distribute.experimental.MultiWorkerMirroredStrategy` и `tf.distribute.TPUStrategy`. Размещение цикла внутри `tf.function` работает для одного рабочего` tf.distribute.TPUStrategy`, но не при использовании TPU модулей."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "zt3AHb46Tr3w"
      },
      "outputs": [],
      "source": [
        "global_batch_size = 16\n",
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "\n",
        "dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(100).batch(global_batch_size)\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)\n",
        "\n",
        "@tf.function\n",
        "def train_step(inputs):\n",
        "  features, labels = inputs\n",
        "  return labels - 0.3 * features\n",
        "\n",
        "for x in dist_dataset:\n",
        "  # метод train_step обучает модель, используя элементы набора данных\n",
        "  loss = mirrored_strategy.run(train_step, args=(x,))\n",
        "  print(\"Loss is \", loss)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "NchPwTEiuSqb"
      },
      "source": [
        "#### Используйте `iter` для создания явного итератора\n",
        "\n",
        "Чтобы перебирать элементы в экземпляре `tf.distribute.DistributedDataset`, вы можете создать `tf.distribute.DistributedIterator`, используя для него `iter`. С помощью явного итератора вы можете выполнять итерацию для фиксированного количества шагов. Чтобы получить следующий элемент из `tf.distribute.DistributedIterator`(экземпляра `dist_iterator`), вы можете вызвать `next(dist_iterator)`, `dist_iterator.get_next()` или `dist_iterator.get_next_as_optional()`. Первые два по сути одинаковы:"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OrMmakq5EqeQ"
      },
      "outputs": [],
      "source": [
        "num_epochs = 10\n",
        "steps_per_epoch = 5\n",
        "for epoch in range(num_epochs):\n",
        "  dist_iterator = iter(dist_dataset)\n",
        "  for step in range(steps_per_epoch):\n",
        "    # метод train_step обучает модель, используя элементы набора данных\n",
        "    loss = mirrored_strategy.run(train_step, args=(next(dist_iterator),))\n",
        "    # args=(next(dist_iterator),) - тоже самое что и args=(dist_iterator.get_next(),)\n",
        "    # в таком вызове\n",
        "    # loss = mirrored_strategy.run(train_step, args=(dist_iterator.get_next(),))\n",
        "    print(\"Loss is \", loss)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "UpJXIlxjqPYg"
      },
      "source": [
        "С помощью `next()` или `tf.distribute.DistributedIterator.get_next()`, если `tf.distribute.DistributedIterator` достиг своего конца, будет выдана ошибка `OutOfRange`. Пользователь может перехватить ошибку на стороне Python и продолжить выполнение других задач, таких как создание контрольных точек и оценка. Однако это не сработает, если вы используете цикл обучения хоста(т.е. Выполняете несколько шагов для каждой `tf.function`), который выглядит так:\n",
        "\n",
        "```\n",
        "@tf.function\n",
        "def train_fn(iterator):\n",
        "  for _ in tf.range(steps_per_loop):\n",
        "    strategy.run(step_fn, args=(next(iterator),))\n",
        "```\n",
        "\n",
        " `train_fn` содержит несколько шагов, заключенных в `tf.range`. В этом случае разные итерации цикла могут запускаться параллельно, и ошибка `OutOfRange` может быть вызвана на более поздних итерациях до завершения вычисления предыдущих итераций. При возникновении исключения `OutOfRange` все операции внутри функции будут немедленно завершены, а значит будут прерваны все циклы, даже те, которые не полностью завершились. Если вы хотите избежать таких ошибок, альтернативой, которая не вызывает ошибку `OutOfRange`, является `tf.distribute.DistributedIterator.get_next_as_optional()`. `get_next_as_optional` возвращает `tf.experimental.Optional`, который содержит следующий элемент или не имеет значения(None), если `tf.distribute.DistributedIterator` достиг своего конца. "
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Iyjao96Vqwyz"
      },
      "outputs": [],
      "source": [
        "# Вы можете прервать цикл с помощью get_next_as_optional, проверив, содержит ли Optional значение\n",
        "global_batch_size = 4\n",
        "steps_per_loop = 5\n",
        "strategy = tf.distribute.MirroredStrategy(devices=[\"GPU:0\", \"CPU:0\"])\n",
        "\n",
        "dataset = tf.data.Dataset.range(9).batch(global_batch_size)\n",
        "distributed_iterator = iter(strategy.experimental_distribute_dataset(dataset))\n",
        "\n",
        "@tf.function\n",
        "def train_fn(distributed_iterator):\n",
        "  for _ in tf.range(steps_per_loop):\n",
        "    optional_data = distributed_iterator.get_next_as_optional()\n",
        "    if not optional_data.has_value():\n",
        "      break\n",
        "    per_replica_results = strategy.run(lambda x:x, args=(optional_data.get_value(),))\n",
        "    tf.print(strategy.experimental_local_results(per_replica_results))\n",
        "train_fn(distributed_iterator)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "LaclbKnqzLjf"
      },
      "source": [
        "## Использование свойства `element_spec` "
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Z1YvXqOpwy08"
      },
      "source": [
        "Если вы передаете элементы распределенного набора данных в `tf.function` и хотите получить `tf.TypeSpec`, вы можете указать аргумент `input_signature` функции `tf.function`. Выходные данные распределенного датасета - это `tf.distribute.DistributedValues`, которые могут представлять входные данные для одного или нескольких устройств. Чтобы получить `tf.TypeSpec`, соответствующий этому распределенному значению `tf.distribute.DistributedValues`, вы можете использовать свойство `element_spec` распределенного датасета или распределенного итератора."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "pg3B-Cw_cn3a"
      },
      "outputs": [],
      "source": [
        "global_batch_size = 16\n",
        "epochs = 5\n",
        "steps_per_epoch = 5\n",
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "\n",
        "dataset = tf.data.Dataset.from_tensors(([1.],[1.])).repeat(100).batch(global_batch_size)\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)\n",
        "\n",
        "@tf.function(input_signature=[dist_dataset.element_spec])\n",
        "def train_step(per_replica_inputs):\n",
        "  def step_fn(inputs):\n",
        "    return 2 * inputs\n",
        "\n",
        "  return mirrored_strategy.run(step_fn, args=(per_replica_inputs,))\n",
        "\n",
        "for _ in range(epochs):\n",
        "  iterator = iter(dist_dataset)\n",
        "  for _ in range(steps_per_epoch):\n",
        "    output = train_step(next(iterator))\n",
        "    tf.print(output)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-OAa6svUzuWm"
      },
      "source": [
        "## Частичные пакеты"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hW2_gVkiztUG"
      },
      "source": [
        "Частичные пакеты возникают, когда экземпляры `tf.data.Dataset`, создаваемые пользователями, могут содержать размеры пакетов, которые не делятся без остатка на количество реплик, или когда размер всех данных не делится на размер пакета. В этом случае, когда набор данных распределяется по нескольким репликам, вызов `next` на некоторых итераторах приведет к ошибке `OutOfRangeError`. Для обработки такой ситуации `tf.distribute` возвращает фиктивные пакеты с размером 0 на репликах, у которых больше нет данных для обработки."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "rqutdpqtPcCH"
      },
      "source": [
        "Для единичного воркера, если данные не получены вызовом `next` на итераторе, создаются фиктивные пакеты с размером 0, которые используются вместе с реальными данными. В случае частичных пакетов последний глобальный пакет данных будет содержать реальные данные наряду с фиктивными пакетами данных. Обработчик данных теперь проверяет, есть ли данные у какой-либо из реплик. Если данных нет ни на одной из реплик, выдается ошибка `OutOfRange`.\n",
        "\n",
        "Для случая с несколькими воркерам булево значение, показывающие наличие данных по каждому воркеру, агрегируется с использованием перекрестной связи реплик, и это способ используется, чтобы определить, все ли воркеры завершили обработку распределенного датасета. Поскольку это связано с взаимодействием между воркерам, возникает некоторое снижение производительности."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vehLsgljz90Y"
      },
      "source": [
        "## Важно!"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Nx4jyN_Az-Dy"
      },
      "source": [
        "* При использовании `tf.distribute.Strategy.experimental_distribute_dataset` с несколькими воркерами, пользователи передают `tf.data.Dataset`, который читает из файлов. Если для `tf.data.experimental.AutoShardPolicy` установлено значение `AUTO` или `FILE`, фактический размер пакета для одного шага может быть меньше, чем глобальный размер пакета, определенный пользователем. Это может произойти, когда в файле остается меньше записей чем глобальный размер пакета. Пользователи могут исчерпать набор данных вне зависимости от количества выполняемых шагов или установить для `tf.data.experimental.AutoShardPolicy` значение `DATA`, чтобы избежать этого.\n",
        "\n",
        "* Преобразования набора данных с отслеживанием состояния в настоящее время не поддерживаются в `tf.distribute`, и любые операции с отслеживанием состояния датасета, в настоящее время игнорируются. Например, если в вашем наборе данных есть `map_fn`, которая использует` tf.random.uniform` для поворота изображения, тогда у вас есть граф набора данных, который зависит от состояния(т.е. случайного начального числа) на локальном компьютере, на котором выполняется процесс Python.\n",
        "\n",
        "* Экспериментальные опции `tf.data.experimental.OptimizationOptions`, которые отключены по умолчанию, могут в определенных случаях - например, при использовании вместе с` tf.distribute` - вызывать снижение производительности. Их следует включать только после того, как вы убедитесь, что они повышают производительность вашей сети при распределенном обучении.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dAC_vRmJyzrB"
      },
      "source": [
        "* Порядок, в котором данные обрабатываются воркерам при использовании `tf.distribute.experimental_distribute_dataset` или `tf.distribute.experimental_distribute_datasets_from_function`, не гарантируется. Обычно порядок выполнения требуется, если вы используете `tf.distribute` для масштабирования прогноза. В таком случае вы можете вставить индекс для каждого элемента в пакете и соответственно упорядочить выходные данные. Следующий фрагмент - пример того, как сортировать выходные данные.\n",
        "\n",
        "Примечание: в данном случае удобно использовать `tf.distribute.MirroredStrategy()`. Нам нужно изменить порядок входных данных когда мы используем несколько воркеров, а `tf.distribute.MirroredStrategy` используется для обучения по одному воркеру."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Zr2xAy-uZZaL"
      },
      "outputs": [],
      "source": [
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "dataset_size = 24\n",
        "batch_size = 6\n",
        "dataset = tf.data.Dataset.range(dataset_size).enumerate().batch(batch_size)\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)\n",
        "\n",
        "def predict(index, inputs):\n",
        "  outputs = 2 * inputs\n",
        "  return index, outputs\n",
        "\n",
        "result = {}\n",
        "for index, inputs in dist_dataset:\n",
        "  output_index, outputs = mirrored_strategy.run(predict, args=(index, inputs))\n",
        "  indices = list(mirrored_strategy.experimental_local_results(output_index))\n",
        "  rindices = []\n",
        "  for a in indices:\n",
        "    rindices.extend(a.numpy())\n",
        "  outputs = list(mirrored_strategy.experimental_local_results(outputs))\n",
        "  routputs = []\n",
        "  for a in outputs:\n",
        "    routputs.extend(a.numpy())\n",
        "  for i, value in zip(rindices, routputs):\n",
        "    result[i] = value\n",
        "\n",
        "print(result)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "33100a66fe91"
      },
      "source": [
        "<a name=\"tensorinputs\">"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "nNbn7HXx0YqB"
      },
      "source": [
        "### Как мне распределять свои данные, если я не использую tf.data.Dataset?"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dymZixqo0nKK"
      },
      "source": [
        "Иногда пользователи не могут использовать `tf.data.Dataset` для ввода и вышеупомянутые API для распределния датасета на несколько устройств.\n",
        "В таких случаях вы можете использовать необработанные тензоры или входные данные из генератора.\n",
        "\n",
        "### Используйте experimental_distribute_values_from_function для произвольных тензорных входов\n",
        "`strategy.run` принимает` tf.distribute.DistributedValues`, который является возвращаемым значением\n",
        "`next(iterator)`. Чтобы передать значения тензора, используйте `experimental_distribute_values_from_function` для создания `tf.distribute.DistributedValues` из необработанных тензоров."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "ajZHNRQs0kqm"
      },
      "outputs": [],
      "source": [
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "worker_devices = mirrored_strategy.extended.worker_devices\n",
        "\n",
        "def value_fn(ctx):\n",
        "  return tf.constant(1.0)\n",
        "\n",
        "distributed_values = mirrored_strategy.experimental_distribute_values_from_function(value_fn)\n",
        "for _ in range(4):\n",
        "  result = mirrored_strategy.run(lambda x:x, args=(distributed_values,))\n",
        "  print(result)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "P98aFQGf0x_7"
      },
      "source": [
        "### Используйте tf.data.Dataset.from_generator если вам нужен вход из генератора"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "emZCWQSi04qT"
      },
      "source": [
        "Если у вас есть функция-генератор, которую вы хотите использовать, вы можете создать экземпляр `tf.data.Dataset` с помощью метода `from_generator`.\n",
        "\n",
        "Примечание. В настоящее время этот метод не поддерживается для `tf.distribute.TPUStrategy`."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "jRhU0X230787"
      },
      "outputs": [],
      "source": [
        "mirrored_strategy = tf.distribute.MirroredStrategy()\n",
        "def input_gen():\n",
        "  while True:\n",
        "    yield np.random.rand(4)\n",
        "\n",
        "# используйте Dataset.from_generator\n",
        "dataset = tf.data.Dataset.from_generator(\n",
        "    input_gen, output_types=(tf.float32), output_shapes=tf.TensorShape([4]))\n",
        "dist_dataset = mirrored_strategy.experimental_distribute_dataset(dataset)\n",
        "iterator = iter(dist_dataset)\n",
        "for _ in range(4):\n",
        "  mirrored_strategy.run(lambda x:x, args=(next(iterator),))"
      ]
    }
  ],
  "metadata": {
    "colab": {
      "collapsed_sections": [],
      "name": "input.ipynb",
      "toc_visible": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}
